package day04;

import data.ClickSource;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import po.Event;

import java.sql.Timestamp;

/**
 * @Description: 每个用户每5秒钟窗口的pv
 *              增量聚合函数和全窗口聚合函数结合使用
 * @Author: ZYX
 * @Date: 2022/2/11 11:00
 * @Version: 1.0
 */
public class Demo02 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        env.addSource(new ClickSource())
                .keyBy(obj -> obj.user)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .aggregate(new CountAgg(),new WindowResult())
                .print();

        env.execute();
    }

    // 输入的泛型是增量聚合函数的输出的类型
    private static class WindowResult extends ProcessWindowFunction<Integer,String,String, TimeWindow> {
        @Override
        public void process(String key, Context context, Iterable<Integer> elements, Collector<String> out) throws Exception {
            // 在窗口关闭的时候，触发调用
            // 迭代器参数中只包含一个元素，就是增量聚合函数发送过来的聚合结果
            long start = context.window().getStart();
            long end = context.window().getEnd();
            Integer next = elements.iterator().next();
            out.collect("用户：" + key + " 在窗口" +
                    "" + new Timestamp(start) + "~" + new Timestamp(end) + "" +
                    "中的pv次数是：" + next);
        }
    }

    public static class CountAgg implements AggregateFunction<Event,Integer,Integer> {
        // 创建累加器
        @Override
        public Integer createAccumulator() {
            return 0;
        }

        // 自定义累加器规则
        @Override
        public Integer add(Event event, Integer account) {

            return account+1;
        }

        // 窗口关闭时返回结果
        @Override
        public Integer getResult(Integer integer) {
            return integer;
        }

        @Override
        public Integer merge(Integer integer, Integer acc1) {
            return null;
        }
    }


}
